package flink_p1

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.util.FileUtils

import java.io.File
import scala.collection.mutable

object FlinkTest_22_table1_cacheFile {

  /**
   * 关联维度表：cachefile
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val FILE_NAME = "id2city"

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.registerCachedFile("data/id2city.txt", FILE_NAME)

    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream.map(new RichMapFunction[String, (String, String)] {

      val idCityMap = new mutable.HashMap[String, String]()

      override def open(parameters: Configuration): Unit = {

        val filename: File = getRuntimeContext.getDistributedCache.getFile(FILE_NAME)
        val strings: Array[String] = FileUtils.readFileUtf8(filename).split("\n")
        for (elem <- strings) {
          val kv: Array[String] = elem.split(" ")
          idCityMap.put(kv(0), kv(1))
        }
      }

      override def map(value: String): (String, String) = {
        (value, idCityMap.getOrElse(value, "null"))
      }
    }).print()

    env.execute()
  }

}
